package com.caissa.springboot.starter.kafka.service;


import com.alibaba.fastjson.JSONObject;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Message;
import com.dianping.cat.message.Transaction;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * kafka 二次封装工具类
 *
 * @author yuzhongo
 * @date 2020/8/19 PM5:45
 */
@Slf4j
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 获取 kafkaTemplate
     *
     * @return
     */
    public KafkaTemplate<String, Object> getKafkaTemplate() {
        return kafkaTemplate;
    }

    /**
     * 异步发送消息
     *
     * @param topic
     * @param object
     */
    public ListenableFuture<SendResult<String, Object>> send(String topic, Object object) {
        return sendAsyncWay(topic, object);
    }

    /**
     * 同步发送消息
     * @param topic
     * @param object
     * @return
     * @throws Exception  5秒不返回报超时异常
     */
    public SendResult<String, Object> sendSyncWay(String topic, Object object) throws Exception {
        return sendSyncWay(topic, object, 5L, TimeUnit.SECONDS);
    }

    private ListenableFuture<SendResult<String, Object>> sendAsyncWay(String topic, Object object) {
        Objects.requireNonNull(topic);
        Objects.requireNonNull(object);
        Transaction kafkaTrans = Cat.newTransaction("kafka", new StringBuffer("producer::sendAsyncWay::").append(topic).toString());
        try {
            ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(topic, JSONObject.toJSONString(object));
            kafkaTrans.setStatus(Message.SUCCESS);
            return listenableFuture;
        } catch (Exception e) {
            kafkaTrans.setStatus(e);
            throw e;
        } finally {
            kafkaTrans.complete();
        }
    }

    /**
     * 同步发送消息
     * @param topic
     * @param object
     * @param timeout
     * @param unit
     * @return
     * @throws Exception
     */
    private SendResult<String, Object> sendSyncWay(String topic, Object object,Long timeout, TimeUnit unit) throws Exception {
        Objects.requireNonNull(topic);
        Objects.requireNonNull(object);
        Objects.requireNonNull(timeout);
        Objects.requireNonNull(unit);
        Transaction kafkaTrans = Cat.newTransaction("kafka", new StringBuffer("producer::sendSyncWay::").append(topic).toString());
        try {
            ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(topic, JSONObject.toJSONString(object));
            SendResult<String, Object> sendResult = listenableFuture.get(timeout, unit);
            kafkaTrans.setStatus(Message.SUCCESS);
            return sendResult;
        } catch (Exception e) {
            kafkaTrans.setStatus(e);
            throw e;
        }finally {
            kafkaTrans.complete();
        }
    }
}
